package org.sn.jdish.spark.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.DataFrameWriter;
import org.apache.spark.sql.SQLContext;

/**
 * format()简单格式的格式转换存储
 * 
 * @author snzigod@hotmail.com
 */
public class SpecifyFormatLoadSave {

	public static void main(String[] args) {
		/**
		 * SparkConf:第一步创建一个SparkConf，在这个对象里面可以设置允许模式Local Standalone yarn
		 * AppName(可以在Web UI中看到) 还可以设置Spark运行时的资源要求
		 */
		SparkConf sparkConf = new SparkConf().setAppName("sparkSql").setMaster("local");

		/**
		 * 基于SparkConf的对象可以创建出来一个SparkContext Spark上下文
		 * SparkContext是通往集群的唯一通道，SparkContext在创建的时候还会创建任务调度器
		 */
		JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

		readJson(javaSparkContext);

		javaSparkContext.close();
	}

	public static void readJson(JavaSparkContext javaSparkContext) {
		// 创建DataFrame 读取json
		SQLContext sqlContext = new SQLContext(javaSparkContext);

		DataFrameReader dataFrameReader = sqlContext.read();

		// parquet 是本地数据存储的格式
		DataFrame df = dataFrameReader.format("json").load("data/people.json");

		// 通过关writer写入并保存save
		DataFrameWriter write = df.select("name").write();
		write.format("parquet").save("people.parquet");
	}
}
